package tk.xboot.flink.kfk.wordcount;

//import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;

import com.sun.management.OperatingSystemMXBean;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.*;

import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.util.Properties;
import java.util.Random;

/**
 * Generate the data sink to kafka
 */
public class DataProducer {

    //private static final ObjectMapper om = new ObjectMapper().registerModule(new JavaTimeModule());
    private static final ObjectMapper om = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        InputStream in = DataProducer.class.getClassLoader().getResourceAsStream("kfk-producer.properties");
        props.load(in);
        produceMemoryUsageStringMessage(props);

    }
    public static void produceMemoryUsageStringMessage(Properties props) {
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            int totalMessageCount = 100000;
            Random r = new Random();

            for (int i = 0; i < totalMessageCount; i++) {
                int m = Math.abs(r.nextInt() % 17) + 1;
                int rdm = i % m;
                long cur = System.currentTimeMillis();
                final int x = i;
                String value = String.format("%d,%s%d,%d", cur, "machine-", rdm, currentMemSize()/1024);
                producer.send(new ProducerRecord<>(props.getProperty("topic"), /*null, cur, rdm + "",*/ value), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.out.println("Failed to send message with exception " + exception);
                        } else {
                            System.out.printf("Send data:{idx:[%d] value:[%s]}\n",x,value);
                        }
                    }
                });
                Thread.sleep(1000L);
            }
            //producer.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class MemoryUsageExtrator {
        private static OperatingSystemMXBean mxBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();

        /**
         * Get current free memory size in bytes
         *
         * @return free RAM size
         */
        public static long currentFreeMemorySizeInBytes() {
            return mxBean.getFreePhysicalMemorySize();
        }
    }

    private static long currentMemSize() {
        return MemoryUsageExtrator.currentFreeMemorySizeInBytes();
    }
}